package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.exception.CustomException;
import com.heima.common.redis.CacheService;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;

@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    TaskinfoMapper taskinfoMapper;
    @Autowired
    TaskinfoLogsMapper taskinfoLogsMapper;
    @Autowired
    CacheService cacheService;

    //定期将db中数据 同步到redis
    @Scheduled(cron = "0 */5 * * * ?")//每五分钟执行一次
    @PostConstruct//项目启动时执行该方法
    public void reloadDbTaskToRedis(){
        String lock = cacheService.tryLock("FUTURE_TASK_DBTOREDIS", 30000);//获取分布式锁，30秒
        if(StringUtils.isNotBlank(lock)){
            log.info("定期将db中数据 同步到redis任务执行了   获取到了分布式锁，当前节点执行任务");
            //查询db中未来5分钟内的任务  同步redis
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE, 5);
            Date time = calendar.getTime();//未来5分钟的时间点
            //select * from taskinfo where executetime < 未来5分钟的时间点 and in_redis=0
            List<Taskinfo> taskinfos = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery()
                    .lt(Taskinfo::getExecuteTime, time)
                    .eq(Taskinfo::getInRedis, 0));
            //存入redis
            if(taskinfos!=null && taskinfos.size()>0){
                for (Taskinfo taskinfo : taskinfos) {
                    Task task = new Task();
                    BeanUtils.copyProperties(taskinfo, task);
                    task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                    addTaskToRedis(task);//添加未来5分钟内的任务到redis
                }
            }
        }else{
            log.info("定期将db中数据 同步到redis任务执行了   但是没有获取到分布式锁，当前节点不执行任务");
        }
    }

    //未来数据队列zset   定时刷新到list结构
    @Scheduled(cron = "0 * * * * ?")//测试阶段每分钟执行一次   正常情况下是每秒执行一次
    public void refresh(){
        String lock = cacheService.tryLock("FUTURE_TASK", 30000);//获取分布式锁，30秒
        if(StringUtils.isNotBlank(lock)){
            log.info("未来数据队列zset定时刷新到list结构任务执行了   获取到了分布式锁，当前节点执行任务");
            Set<String> scan = cacheService.scan(ScheduleConstants.FUTURE + "*");//future_*
            //future_1001_1      future_1002_1       future_1003_1
            for (String futureKey : scan) {
                Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());//从未来数据队列中获取任务  获取到任务则说明该任务执行时间小于当前时间，该执行了
                if(tasks!=null && tasks.size()>0){
                    //future_1001_1      future_1002_1       future_1003_1
                    //topic_1001_1       topic_1002_1        topic_1003_1
                    String topicKey = futureKey.replace(ScheduleConstants.FUTURE, ScheduleConstants.TOPIC);
                    cacheService.refreshWithPipeline(futureKey, topicKey, tasks);//开启管道将zset数据分批刷到 list结构中
                    log.info("成功将 {} 中的数据 刷新到 {}  任务数量为{}", futureKey, topicKey, tasks.size());
                }
            }
        }else{
            log.info("未来数据队列zset定时刷新到list结构任务执行了   但是没有获取到分布式锁，当前节点不执行任务");
        }
    }




    @Override
    public Task pull(int type, int priority) {
        //1.校验
        if(type<=0 || priority<=0){
            return null;
        }

        Task task = null;
        //2.从redis的list结构中拉取   poll命令
        String key = type+"_"+priority;// 1001_1 充话费     1002_1 充游戏点卡
        String pop = cacheService.lRightPop(ScheduleConstants.TOPIC + key);//从list结构弹出元素
        if(StringUtils.isNotBlank(pop)){
            //3.如果拉取到任务id则 从hash结构中获取任务详情
            String taskJSONStr = (String)cacheService.hGet("hash_" + key, pop);
            task = JSON.parseObject(taskJSONStr, Task.class);

            //4.修改db中taskinfo和taskinfo_log表数据
            updateTask(task.getTaskId(), ScheduleConstants.EXECUTED);
        }
        return task;
    }

    @Override
    public boolean cancelTask(long taskId) {
        //1.校验
        if(taskId<=0){
            return false;
        }

        //2.修改db   删除taskinfo表数据  taskinfo_log表status改成2 version自增
        Task task = updateTask(taskId, ScheduleConstants.CANCELLED);

        //3.删除redis中的任务数据  任务执行时间小于等于当前时间去list中删除   小于未来5分钟时间点去zset中删除
        removeTaskFromRedis(task);
        return true;
    }

    /**
     * 从redis中删除 任务    任务执行时间小于等于当前时间去list中删除   小于未来5分钟时间点去zset中删除
     * @param task
     */
    private void removeTaskFromRedis(Task task) {
        String key = task.getTaskType()+"_"+task.getPriority();// 1001_1 充话费     1002_1 充游戏点卡
        if(task.getExecuteTime()<=System.currentTimeMillis()){
            //从list中删除
            cacheService.lRemove(ScheduleConstants.TOPIC+key, 0, task.getTaskId().toString());
        }else{
            //从zset中删除
            cacheService.zRemove(ScheduleConstants.FUTURE+key, task.getTaskId().toString());
        }
        //删除hash中的任务数据
        cacheService.hDelete("hash_"+key,  task.getTaskId().toString());
    }

    /**
     * 修改任务状态    删除taskinfo表数据  taskinfo_log表status   version自增
     * @param taskId
     * @param status  0初始化   1已执行   2已取消
     * @return
     */
    private Task updateTask(long taskId, int status) {
        taskinfoMapper.deleteById(taskId);//删除任务表数据

        TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
        taskinfoLogs.setStatus(status);
        int update = taskinfoLogsMapper.updateById(taskinfoLogs);//修改时会比较版本号，如果不一致则修改失败 返回的影响行数为0
        if(update==0){
            throw new CustomException(AppHttpCodeEnum.PARAM_INVALID);
        }

        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogs, task);
        task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        return task;
    }

    @Override
    public long addTask(Task task) {
        //1.校验
        if(task==null){
            return 0;
        }
        //2.添加任务到db
        Long taskId = addTaskToDB(task);

        //3.添加任务到缓存
        task.setTaskId(taskId);
        addTaskToRedis(task);
        return taskId;
    }

    /**
     * 添加任务到redis
     * @param task
     */
    private void addTaskToRedis(Task task) {
        //判断任务执行时间
        // 小于等于当前时间存入redis的list结构
        // 大于当前时间小于未来5分钟时间点（未来5分钟内执行的任务）存入zset结构
        // 大于未来5分钟时间点（未来5分钟后执行的任务）   不存redis
        // 将任务数据存入hash结构，未来取任务时从hash结构中取任务，不查db
        String key = task.getTaskType()+"_"+task.getPriority();// 1001_1 充话费     1002_1 充游戏点卡

        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        long timeInMillis = calendar.getTimeInMillis();//未来5分钟时间点  毫秒值

        boolean flag=false;//用于判断是否存redis  hash结构中

        if(task.getExecuteTime()<=System.currentTimeMillis()){
            //立可执行的任务存入list结构     topic_1001_1 充话费    topic_1002_1 充游戏点卡      topic_1003_1 文章发布审核
            cacheService.lLeftPush(ScheduleConstants.TOPIC+key, task.getTaskId().toString());
            flag=true;
        }else if(task.getExecuteTime()< timeInMillis){
            //未来5分钟内执行的任务存入zset结果    future_1001_1 充话费     future_1002_1 充游戏点卡    future_1003_1 文章发布审核
            cacheService.zAdd(ScheduleConstants.FUTURE+key,  task.getTaskId().toString(), task.getExecuteTime());
            flag=true;
        }else{
            log.info("任务是未来5分钟后执行的，先不存redis");
        }

        //存入hash    hash_1001_1 充话费     hash_1002_1 充游戏点卡   hash_1003_1 文章发布审核
        if(flag){
            cacheService.hPut("hash_"+key,  task.getTaskId().toString(), JSON.toJSONString(task));
        }
        //将taskinfo表中in_redis字段修改为1  代表数据在redis中
        taskinfoMapper.update(null, Wrappers.<Taskinfo>lambdaUpdate()
                            .set(Taskinfo::getInRedis,1)
                            .eq(Taskinfo::getTaskId, task.getTaskId()));
    }

    /**
     * 添加任务到db中
     * @param task
     * @return
     */
    private Long addTaskToDB(Task task) {
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskinfoMapper.insert(taskinfo);//保存任务

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo, taskinfoLogs);
        taskinfoLogs.setVersion(1);//默认版本号为1
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);//0初始化  1已执行   2已取消
        taskinfoLogsMapper.insert(taskinfoLogs);
        return taskinfo.getTaskId();
    }
}
